


/*
    Async and parallel processing with service broker
*/
if not exists( select 1 from sys.databases where name = 'AsyncProcessing')
  begin
  create database AsyncProcessing
  end
go
use AsyncProcessing
go
alter database AsyncProcessing set enable_broker
go
if exists (select 1 from sys.objects where name = 'FileContent')
  drop table FileContent
go

create table FileContent
  (fileId int identity(1,1) primary key nonclustered
  ,filename nvarchar(max)
  ,filenameHash int
  ,content varbinary(max) 
  ,dateUploaded datetime default (getdate())
  )
go
create index IX_FileContent on FileContent (filenamehash)
go
create clustered index IXC_FileContent_DateUploaded on FileContent (DateUploaded)
go
if exists (select 1 from sys.objects where name = 'FileUpload')
  drop procedure FileUpload 
go
create procedure FileUpload 
  @filename nvarchar(max)
as
declare @sql nvarchar(max)

  if exists (select 1 from FileContent where filenamehash = checksum(@filename) and filename = @filename)
    delete from FileContent where filenamehash = checksum(@filename) and filename = @filename

  set @sql = '
  insert into FileContent (filename, content)
    select @filename, content
     from openrowset (bulk ''' + replace(@filename,'''','''''') + ''', single_blob) as data(content)'

  exec sp_executesql @sql, N'@filename nvarchar(max)',@filename

go
/* 
  Test the upload
*/
exec FileUpload 'c:\results.csv'
go
select top 1 * from fileContent
go
/*
  clear service broker objects
*/
if exists (select 1 from sys.services where name = 'FileUploadRequest')
  drop service FileUploadRequest
if exists (select 1 from sys.services where name = 'FileUploadResponse')
  drop service FileUploadResponse
if exists (select 1 from sys.service_queues where name = 'FileUploadResponseQueue')
  drop queue FileUploadResponseQueue
if exists (select 1 from sys.service_queues where name = 'FileUploadRequestQueue')
  drop queue FileUploadRequestQueue
go
/*
  Create a bucket to hold requests for file uploads 
*/
create queue FileUploadRequestQueue
go
/*
  Create a bucket to hold the responses returned following a file upload request being actioned
*/
create queue FileUploadResponseQueue
go
/*
  Create a service that defines we are going to accept request messages of any format (DEFAULT contract)
  and store then in the queue we've just created
*/
create service FileUploadRequest
on queue FileUploadRequestQueue
([DEFAULT])
go
/*
  Create another service that defines we are going to accept response messages of any format (DEFAULT contract)
  and store then in the queue we've just created
*/
create service FileUploadResponse
on queue FileUploadResponseQueue
([DEFAULT])
go
/* 
  Now we want to create a procedure that receives the request to upload a file, uploads the file and sends a response back
*/
if exists (select 1 from sys.objects where name = 'FileUploadAsyncWorker')
  drop procedure FileUploadAsyncWorker
go
create procedure FileUploadAsyncWorker
as
  declare @handle uniqueidentifier;
  declare @message nvarchar(max);
  declare @messageType nvarchar(max);
  
  declare @continue int
  set @continue = 1

  waitfor delay '00:00:02'
  
  while @continue = 1
    begin
    --We need to read a message from a queue, this can return multiple messages and so we specify top (1)
    --Messages are implicityl received in the order they were sent, so you don't have to specify an ORDER BY (you can't anyway)
    waitfor (receive top (1) @message = message_body
                            ,@messageType = message_type_name
                            ,@handle = conversation_handle
                 from FileUploadRequestQueue
            ), timeout 500;

    --We should check that we have received a message, if the waitfor had timed out then the rowcount would be 0
    if @@rowcount > 0
      begin
    
      --The type of message that has been recieved needs to be checked
      if @messageType like '%end%dialog%'
        end conversation @handle;
      
      if @messageType = 'DEFAULT'
        begin
        exec FileUpload @message;
        send on conversation @handle (@message + '-Uploaded');
        end

      end
    else 
      set @continue = 0
    end
go

/*
  Enable the activation on the queue
*/
alter queue FileUploadRequestQueue
with activation (status = ON
                ,procedure_name = FileUploadAsyncWorker
                ,max_queue_readers = 10
                ,execute as self
                )
go
/*
  Create the procedure to request an asynchronous upload of a file
*/
if exists (select 1 from sys.objects where name = 'FileUploadAsync')
  drop procedure FileUploadAsync
go
create procedure FileUploadAsync
  @filename nvarchar(max)
 ,@handle uniqueidentifier output
as  
  begin dialog conversation @handle 
  from service FileUploadResponse
  to service 'FileUploadRequest';

  send on conversation @handle (@filename);
  
go
/*
  A procedure to monitor the for a response to the request to upload a file
*/
if exists (select 1 from sys.objects where name = 'FileUploadMonitor')
  drop procedure FileUploadMonitor
go
create procedure FileUploadMonitor
  @handle uniqueidentifier
 ,@timeout int
 ,@message nvarchar(max) output
 as

  declare @messageType nvarchar(max);

  
  waitfor (receive top (1) @message = message_body
                          ,@messageType = message_type_name
               from FileUploadResponseQueue
              where conversation_handle = @handle 
          ), timeout @timeout;

  if @@rowcount > 0
    begin
    end conversation @handle;
    end
go
/*
  Test the queuing with TSQL
*/
declare @handle uniqueidentifier
declare @message nvarchar(max)

truncate table FileContent
select filename from FileContent

--Change the filename to a file that exists on your system
exec FileUploadAsync 'c:\results.csv',@handle output

Print 'Uploaded while now waiting'

exec FileUploadMonitor @handle,60000,@message output

print @message

select filename from FileContent